package com.algo.rpc.client;

import com.algo.rpc.common.RpcDecoderHandler;
import com.algo.rpc.common.RpcEncoderHandler;
import com.algo.rpc.common.RpcRequest;
import com.algo.rpc.common.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.log4j.Logger;

import java.util.Date;

/**
 * RPC客户端
 */
public class RpcNettyClient {

    private static final Logger logger = Logger.getLogger(RpcNettyClient.class);

    private EventLoopGroup workGroup;

    private Bootstrap bootstrap;

    private ChannelFuture channelFuture;

    private RpcNettyClientHandler clientHandler;

    private RpcRequestTimeoutHandler timeoutHandler;

    private RpcRequestTimeHolder requestTimeHolder = new RpcRequestTimeHolder();

    /**
     * 网络连接
     * @throws InterruptedException
     */
    public void connect(RpcClientConfig rpcClientConfig) throws InterruptedException {
        workGroup = new NioEventLoopGroup(8);
        clientHandler = new RpcNettyClientHandler(requestTimeHolder, rpcClientConfig.getTimeout());
        timeoutHandler = new RpcRequestTimeoutHandler();
        bootstrap = new Bootstrap();

        bootstrap.group(workGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline()
                                .addLast(new RpcDecoderHandler())
                                .addLast(new RpcEncoderHandler())
                                .addLast(timeoutHandler)
                                .addLast(clientHandler);
                    }
                })
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.SO_BACKLOG, 128);
        try {
            channelFuture = bootstrap.connect(rpcClientConfig.getHostName(), rpcClientConfig.getPort()).sync();
            logger.info("rpc client connect server success, hostname: "
                    + rpcClientConfig.getHostName() + ", port: " + rpcClientConfig.getPort());
        } catch (Exception e) {
            logger.error("rpc client connect server error: ", e);
        }
    }

    /**
     * 远程方法调用
     * @param request
     * @return
     */
    public RpcResponse remoteCall(RpcRequest request) throws Throwable {
        logger.info("remote call " + request);
        requestTimeHolder.put(request.getRequestId(), System.currentTimeMillis());
        // 发送请求报文
        channelFuture.channel().writeAndFlush(request).sync();
        // 同步等待请求结果
        RpcResponse response;
        try {
            response = clientHandler.getResposne(request.getRequestId());
            logger.info("remote result is " + response);
        } catch (Throwable throwable) {
            throw throwable;
        }
        return response;
    }

}
